package com.hzgj.bcl.soa.rpc.client;

import com.hzgj.bcl.concurrent.FutureListener;
import com.hzgj.bcl.soa.Address;
import com.hzgj.bcl.soa.rpc.RequestCorrelator;
import com.hzgj.bcl.soa.rpc.RspCollector;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.CondVar;
import com.hzgj.bcl.soa.util.Condition;
import com.hzgj.bcl.soa.util.NotifyingFuture;

import org.doomdark.uuid.UUID;
import org.doomdark.uuid.UUIDGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public abstract class Request implements RspCollector, NotifyingFuture, Condition {
    protected static final Logger log = LoggerFactory.getLogger(Request.class);
    protected static final UUIDGenerator generator = UUIDGenerator.getInstance();
    private static final AtomicLong INVOKE_ID = new AtomicLong(0);
    protected final Lock lock = new ReentrantLock();
    /**
     * Is set as soon as the request has received all required responses
     */
    protected final CondVar cond = new CondVar(lock);
    protected final Message request_msg;
    protected final RequestCorrelator corr;         // either use RequestCorrelator or ...
    protected final RequestOptions options;
    protected final long req_id; // request ID for this request
    protected volatile boolean done;
    protected boolean block_for_results = true;
    protected volatile FutureListener listener;

    public Request(Message request, RequestCorrelator corr, RequestOptions options) {
        this.request_msg = request;
        this.corr = corr;
        this.options = options;
        this.req_id = newId();
    }

    /**
     * Generates a new unique request ID
     */
    protected static String getRequestId() {
        UUID uuid = generator.generateRandomBasedUUID();
        return uuid.toString();
    }

    public static long newId() {
        // getAndIncrement()增长到MAX_VALUE时，再增长会变为MIN_VALUE，负数也可以做为ID
        return INVOKE_ID.getAndIncrement();
    }

    public boolean getBlockForResults() {
        return block_for_results;
    }

    public void setBlockForResults(boolean block_for_results) {
        this.block_for_results = block_for_results;
    }

    @Override
    public NotifyingFuture setListener(FutureListener listener) {
        this.listener = listener;
        if (done) {
            listener.futureDone(this);
        }
        return this;
    }

    public boolean execute() throws Exception {
        if (corr == null) {
            if (log.isErrorEnabled()) {
                log.error("corr is null, cannot handle request");
            }
            return false;
        }

        sendRequest();
        if (!block_for_results) {
            return true;
        }

        lock.lock();
        try {
            //一直阻塞
            return responsesComplete(-1);
        } finally {
            done = true;
            lock.unlock();
        }
    }

    protected abstract void sendRequest() throws Exception;

    @Override
    public abstract void receiveResponse(Object response_value, Address sender, boolean is_exception);

    @Override
    public boolean isMet() {
        return responsesComplete();
    }

    protected abstract boolean responsesComplete();

    public boolean getResponsesComplete() {
        lock.lock();
        try {
            return responsesComplete();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        lock.lock();
        try {
            boolean retval = !done;
            done = true;
            if (corr != null) {
                corr.done(req_id);
            }
            cond.signal(true);
            return retval;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public boolean isCancelled() {
        lock.lock();
        try {
            return done;
        } finally {
            lock.unlock();
        }
    }


    /* --------------------------------- Private Methods -------------------------------------*/

    @Override
    public boolean isDone() {
        return done;
    }

    @Override
    public String toString() {
        StringBuilder ret = new StringBuilder(128);
        ret.append(super.toString());
        ret.append(", req_id=").append(req_id);
        return ret.toString();
    }

    protected void checkCompletion(Future future) {
        if (listener != null && responsesComplete()) {
            listener.futureDone(future);
        }
    }

    /**
     * This method runs with lock locked (called by <code>execute()</code>).
     */
    protected boolean responsesComplete(final long timeout) throws InterruptedException {
        try {
            return waitForResults(timeout);
        } finally {
            if (corr != null) {
                corr.done(req_id);
            }
        }
    }

    protected boolean waitForResults(final long timeout) {
        if (timeout <= 0) {
            cond.waitFor(this);
            return true;
        }
        return cond.waitFor(this, timeout, TimeUnit.MILLISECONDS);
    }


}